package com.yege.email;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.pulsar.client.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.mail.SimpleMailMessage;
import org.springframework.mail.javamail.JavaMailSender;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.Date;

/**
 * @author yege
 */
@Component
public class EmailPulsarConsumer {
    private final static Log log = LogFactory.getLog(EmailPulsarConsumer.class);
    @Value("${pulsar.url}")
    private String url;
    @Value("${pulsar.token}")
    private String token;
    @Value("${pulsar.listenerName}")
    private String listenerName;
    @Value("${pulsar.topic}")
    private String topic;
    @Value("${pulsar.subscription}")
    private String subscription;


    @Value("${active.url}")
    private String activeUrl;
    @Value("${spring.mail.username}")
    private String emailFrom;

    PulsarClient client = null;

    @Autowired
    private JavaMailSender javaMailSender;

    @PostConstruct
    public void initPulsar(){
        try {
            log.debug("initPulsar method was executed once");
            log.info("initPulsar method was executed once");
            log.warn("initPulsar method was executed once");
            log.error("initPulsar method was executed once");


            ClientBuilder clientBuilder =PulsarClient.builder();
            if(!StringUtils.isEmpty(listenerName)){
                clientBuilder.listenerName(listenerName);
            }
            if(!StringUtils.isEmpty(token)){
                clientBuilder.authentication(AuthenticationFactory.token(token));
            }

            //构造Pulsar client
            client = clientBuilder.serviceUrl(url).build();

            Consumer consumer = client.newConsumer()
                    .topic(topic)
                    .subscriptionName(subscription)
                    .subscriptionType(SubscriptionType.Shared)//负载均衡模式(共享模式)
                    //.subscriptionType(SubscriptionType.Exclusive)//独享模式，默认
                    //.subscriptionType(SubscriptionType.Failover)//灾备模式，默认第一个启动的是主，后面启动的实例是备份
                    .subscribe();

            while (true) {
                Message msg = consumer.receive();
                try {
                    // Do something with the message
                    String msgP=new String(msg.getData());
                    String[] msgArr=msgP.split(",");
                    sendMail(msgArr[0],msgArr[1]);
                    log.info("message consume successful: "+msgP);
                    // Acknowledge the message so that it can be deleted by the message broker
                    consumer.acknowledge(msg);
                } catch (Exception e) {
                    log.error(e.getMessage(),e);
                    // Message failed to process, redeliver later
                    consumer.negativeAcknowledge(msg);
                }
            }
        }catch (Exception ex){
            log.error(ex.getMessage(),ex);
        }
    }

    public void sendMail(String toEmail,String code) {
        try {
            log.debug("sendMail method was executed once");
            log.info("sendMail method was executed once");
            log.warn("sendMail method was executed once");
            log.error("sendMail method was executed once");

            // 构建一个邮件对象
            SimpleMailMessage message = new SimpleMailMessage();
            // 设置邮件主题
            message.setSubject("账号激活通知");
            // 设置邮件发送者，这个跟application.yml中设置的要一致
            message.setFrom(emailFrom);
            // 设置邮件接收者，可以有多个接收者，中间用逗号隔开，以下类似
            message.setTo(toEmail);
            // 设置邮件发送日期
            message.setSentDate(new Date());
            // 设置邮件的正文
            message.setText("账号激活地址:"+activeUrl+code);
            // 发送邮件
            javaMailSender.send(message);
            log.info("email send successful,toEmail:"+toEmail);
        }catch (Exception ex){
            log.error(ex.getMessage(),ex);
        }

    }
}